package com.bawei.flink.tableapi;

import org.apache.flink.configuration.Configuration;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.TableEnvironment;

public class TemporalJoinDemo {
    public static void main(String[] args) {
        EnvironmentSettings settings = EnvironmentSettings
                .newInstance()
                .useBlinkPlanner()
                .inStreamingMode()
                .build();

        TableEnvironment tEnv = TableEnvironment.create(settings);
        Configuration config = tEnv.getConfig().getConfiguration();
        config.setString("table.exec.resource.default-parallelism","4");

        tEnv.executeSql("CREATE TABLE currency_rates(\n" +
                "  currency STRING,\n" +
                "  conversion_rate DECIMAL(32, 2),\n" +
                "  update_time BIGINT,\n" +
                "  update_time_ltz AS TO_TIMESTAMP_LTZ(update_time,3),\n" +
                "  WATERMARK FOR update_time_ltz AS update_time_ltz\n" +
                ") WITH (\n" +
                " 'connector' = 'kafka',\n" +
                " 'topic' = 'currency_rates',\n" +
                " 'properties.bootstrap.servers' = 'cdh1:9092,cdh2:9092,cdh3:9092',\n" +
                " 'properties.group.id' = 'testGroup',\n" +
                " 'scan.startup.mode' = 'latest-offset',\n" +
                " 'format' = 'csv',\n" +
                " 'csv.ignore-parse-errors' = 'true',\n" +
                " 'csv.allow-comments' = 'true'\n" +
                ")");

          tEnv.executeSql("CREATE VIEW versioned_rates AS \n" +
                  "SELECT currency, conversion_rate, update_time_ltz FROM (\n" +
                  "SELECT *,ROW_NUMBER() OVER(PARTITION BY currency ORDER BY update_time_ltz DESC) AS rk FROM currency_rates)\n " +
                  "WHERE rk = 1");

        tEnv.executeSql("CREATE TABLE orders(\n" +
                "  order_id    STRING,\n" +
                "  price       DECIMAL(32,2),\n" +
                "  currency    STRING,\n" +
                "  order_time  BIGINT,\n" +
                "  order_time_ltz AS TO_TIMESTAMP_LTZ(order_time,3),\n" +
                "  WATERMARK FOR order_time_ltz AS order_time_ltz\n" +
                ") WITH (\n" +
                " 'connector' = 'kafka',\n" +
                " 'topic' = 'orders',\n" +
                " 'properties.bootstrap.servers' = 'cdh1:9092,cdh2:9092,cdh3:9092',\n" +
                " 'properties.group.id' = 'testGroup',\n" +
                " 'scan.startup.mode' = 'latest-offset',\n" +
                " 'format' = 'csv',\n" +
                " 'csv.ignore-parse-errors' = 'true',\n" +
                " 'csv.allow-comments' = 'true'\n" +
                ")");
//
        //i am shiguowei i have social comm nb
          tEnv.sqlQuery("select order_id,price,orders.currency currency,conversion_rate  from orders left join versioned_rates FOR SYSTEM_TIME AS OF orders.order_time_ltz ON orders.currency = versioned_rates.currency").execute().print();
//          tEnv.sqlQuery("select * from orders").execute().print();
//          tEnv.sqlQuery("select * from versioned_rates").execute().print();

    }
}
